package com.windy.common.transmit.kafka;

import cn.hutool.core.util.ObjectUtil;
import com.windy.common.context.HeaderContextHolder;
import com.windy.common.header.Headers;
import com.windy.common.header.KafkaHeader;
import com.windy.common.transmit.TransmitHeader;

import java.util.function.Supplier;

/**
 * Created by liyufeng on 2022/5/9
 */
public class KafkaProducerTransmitHeader extends TransmitHeader<org.apache.kafka.common.header.Headers> {


    @Override
    protected void replayHeader(Headers headers, Supplier<org.apache.kafka.common.header.Headers> supplier) {

    }

    @Override
    protected Headers pickerHeader(Supplier<org.apache.kafka.common.header.Headers> supplier) {
        org.apache.kafka.common.header.Headers headers = supplier.get();
        Headers context=ObjectUtil.defaultIfNull(HeaderContextHolder.getHeader(),new Headers());
        headers.iterator().forEachRemaining(header -> {
           if(!context.contains(header.key())){
               context.add(new KafkaHeader(header));
           }
        });
        return context;
    }
}
